package com.whz.bus.util;

import com.google.common.collect.Maps;
import io.debezium.connector.mysql.legacy.SourceInfo;
import io.debezium.data.Envelope;
import io.debezium.engine.RecordChangeEvent;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import com.whz.bus.consumer.RelationalChangeData;

import java.util.Map;
import java.util.Optional;
import java.util.function.Supplier;
import java.util.stream.Collectors;

/**
 * @author whz
 */
@Slf4j
@UtilityClass
public class StructParser {
    public Struct getStruct(final Struct source, final String filedName) {
        return execute(() -> source.getStruct(filedName));
    }

    public Long getLong(final Struct source, final String filedName) {
        return execute(() -> source.getInt64(filedName));
    }

    public String getString(final Struct source, final String filedName) {
        return execute(() -> source.getString(filedName));
    }

    public <T> T execute(Supplier<T> supplier) {
        try {
            return supplier.get();
        } catch (Exception exception) {
            log.error("get value from struct异常.", exception);
        }
        return null;
    }


    /**
     * 从debezium封装的record构建java pojo data
     *
     * @param recordRecordChangeEvent debezium 封装的record
     * @return 数据库变化的数据, java 对象
     */
    public RelationalChangeData buildFromDebeziumRecord(RecordChangeEvent<SourceRecord> recordRecordChangeEvent) {
        return Optional.ofNullable(recordRecordChangeEvent)
                .filter(StructParser::isStruct)
                .map(StructParser::convertToStruct)
                .filter(StructParser::whenDataChange)
                .map(StructParser::buildRelationalChangeData)
                .orElse(null);
    }

    private RelationalChangeData buildRelationalChangeData(final Struct source) {
        // 获取增删改对应的结构体数据
        String op = StructParser.getString(source, Envelope.FieldName.OPERATION);

        // 是delete么? delete 操作拿到的是delete前的数据
        Envelope.Operation operation = Envelope.Operation.forCode(op);

        boolean isDeleted = operation == Envelope.Operation.DELETE;
        String sequence = isDeleted ? Envelope.FieldName.BEFORE : Envelope.FieldName.AFTER;
        Struct changeData = StructParser.getStruct(source, sequence);

        /*
         * 参考 LegacyV1MySqlSourceInfoStructMaker
         * 数据库变化原始封装
         */
        Struct origin = StructParser.getStruct(source, Envelope.FieldName.SOURCE);
        // 时间戳
        Long ts = StructParser.getLong(origin, Envelope.FieldName.TIMESTAMP);
        // 数据库名
        String dbName = StructParser.getString(origin, SourceInfo.DATABASE_NAME_KEY);
        // 表名
        String tableName = StructParser.getString(origin, SourceInfo.TABLE_NAME_KEY);

        // 将变更的行数据封装为Map
        Map<String, Object> row = Optional.ofNullable(changeData).map(struct ->
                struct.schema().fields().stream()
                        .filter(field -> !field.schema().isOptional())// 有值
                        .map(field -> Pair.of(field.name(), struct.get(field)))// 字段name,字段value
                        .collect(Collectors.toMap(Pair::getKey, Pair::getValue))
        ).orElse(Maps.newHashMap());

        // 封装
        String opName = Optional.ofNullable(operation).map(Envelope.Operation::name).orElse("unknown op");
        return RelationalChangeData.builder()
                .dbName(dbName)
                .tableName(tableName)
                .row(row)
                .timeStamp(ts)
                .operation(opName)
                .build();
    }

    private Struct convertToStruct(final RecordChangeEvent<SourceRecord> r) {
        log.debug("record value:{}", r.record().value());
        return (Struct) r.record().value();
    }

    private boolean isStruct(final RecordChangeEvent<SourceRecord> r) {
        log.debug("record:{}", r.record());
        return r.record().value() instanceof Struct;
    }

    private boolean whenDataChange(final Struct source) {
        log.debug("record value :{}", source);
        if (source.schema().name().contains("SchemaChange")) {
            // db scheme change
            log.info("暂时不支持scheme change event:{}.", source);
            return false;
        }
        return hasEnvelopeCode(source);
    }

    private boolean hasEnvelopeCode(final Struct source) {
        try {
            return Envelope.Operation.forCode(StructParser.getString(source, Envelope.FieldName.OPERATION)) != null;
        } catch (Exception e) {
            log.warn("hasEnvelopeCode异常.", e);
        }
        return false;
    }
}